package com.lac.acme.kafka;

import org.junit.jupiter.api.Test;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;

public class CompletableFutureTest {
    private Executor poolExecutor = new ThreadPoolExecutor(10, 20, 10000L, TimeUnit.SECONDS, new LinkedBlockingDeque<>());

    @Test
    public void whenApplyTest() throws IOException {
        CompletableFuture<String> start = CompletableFuture.completedFuture("start");
        CompletableFuture<String> cf = start.supplyAsync(() -> {
            for (int i = 0; i < 100; i++) {
                String theadName = Thread.currentThread().getName();
                System.out.println(theadName + ":" + i);
                try {
                    TimeUnit.MILLISECONDS.sleep(50);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            return "task 完成了";
        }, poolExecutor).thenApply((res) -> {
            String theadName = Thread.currentThread().getName();
            System.out.println(theadName + ":" + res);
            return res;
        });

        System.in.read();
    }
    @Test
    public void whenCompleteTest() throws IOException {
        CompletableFuture<String> start = CompletableFuture.completedFuture("start");
        CompletableFuture<String> cf = start.supplyAsync(() -> {
            for (int i = 0; i < 100; i++) {
                String theadName = Thread.currentThread().getName();
                System.out.println(theadName + ":" + i);
                try {
                    TimeUnit.MILLISECONDS.sleep(50);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            return "task 完成了";
        }, poolExecutor).whenComplete((res, exception) -> {
            String theadName = Thread.currentThread().getName();
            System.out.println(theadName + ":" + res);
        });
//        cf.join();
//        String res = cf.getNow("FAIL");
        System.in.read();

//        System.out.println(res);
    }
    @Test
    public void anyOfExampleTest() throws IOException {
        List<String> messages = Arrays.asList("a", "b", "c");
        List<CompletableFuture<String>> fetureList = messages.stream()
                .map(CompletableFuture::completedFuture)
                .map(cf -> cf.thenApply(s -> {
                    try {
                        TimeUnit.SECONDS.sleep(5);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return s.toUpperCase();
                }))
                .collect(Collectors.toList());

        CompletableFuture.allOf(fetureList.toArray(new CompletableFuture[]{})).whenComplete((s, exception) -> {
            System.out.println(s);
        });

        System.out.println("--------end---------");
        System.in.read();

    }

    @Test
    public void thenApplyExample() {
        CompletableFuture<String> cf = CompletableFuture.completedFuture("message").thenApply(s -> {
            System.out.println(Thread.currentThread().isDaemon());
            return s.toUpperCase();
        });
        System.out.println("MESSAGE:"+ cf.getNow(null));
    }
}
